package com.lglbc.rocketmq;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @Author 乐哥聊编程
 * @Doc 关注公众号"乐哥聊编程"获取文档和源码
 * @Date 2023/6/23
 * @Description
 */
@RestController
public class TestController {
    @Autowired
    private StreamBridge streamBridge;

    @RequestMapping("/testSendMsg")
    public String send(String msg){
        Map<String, Object> headers = new HashMap<>();
        headers.put(MessageConst.PROPERTY_TAGS, "tagStr");
        headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, 1);
        Message<String> message = MessageBuilder.createMessage(msg, new MessageHeaders(headers));
        streamBridge.send("producer-out-0",message);
        return JSON.toJSONString(message);
    }

    @RequestMapping("/sendDelayMsg")
    public String sendDelayMsg(String msg){
//      messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
        Map<String, Object> headers = new HashMap<>();
        headers.put(MessageConst.PROPERTY_KEYS, UUID.randomUUID().toString());
        headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, 1);
        // 设置延时等级1~10
        headers.put(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 3);
        Message<String> message = new GenericMessage(msg, headers);
        System.out.println("消息将在30s后被客户端消费："+System.currentTimeMillis()/1000);
        streamBridge.send("producer-out-0",message);
        return JSON.toJSONString(message);
    }

    @RequestMapping("sendOrderMsg")
    public String sendOrderMsg() {
        for (int i = 0; i < 100; i++) {
            String key = "KEY" + i;
            Map<String, Object> headers = new HashMap<>();
            headers.put(MessageConst.PROPERTY_KEYS, key);
            headers.put(MessageConst.PROPERTY_TAGS, Constant.tags[i % Constant.tags.length]);
            headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i);
            Message<String> msg = new GenericMessage(String.format("hello 我是第%s个消息。我的tag为%s",i, Constant.tags[i % Constant.tags.length]), headers);
            streamBridge.send("producer-out-0", msg);
        }
        return "success";
    }
}
